package com.lleice.web.message.kafka.consumer;

import java.util.Optional;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import com.lleice.service.entity.sys.SysRequestLog;
import com.lleice.service.service.sys.ISysRequestLogService;
import com.lleice.web.message.kafka.topic.TopicConstant;
import com.google.gson.Gson;

import lombok.extern.slf4j.Slf4j;

/**
 * 
 * @author lizihao
 *
 */
@Component
@ConditionalOnProperty(prefix = "spring.kafka", name = "enable", havingValue = "true")
@Slf4j
public class KafkaConsumer {

	@Autowired
	private ISysRequestLogService iSysRequestLogService;
	
	@KafkaListener(topics = {TopicConstant.REQUEST_LOG_TOPIC})
    public void requestLogListener(ConsumerRecord<String, String> record){
		try {
			Optional<?> kafkaMessage = Optional.ofNullable(record.value());
	        if (kafkaMessage.isPresent()) {
	            Object message = kafkaMessage.get();   
	            SysRequestLog log = new Gson().fromJson((String)message, SysRequestLog.class);
	            iSysRequestLogService.save(log);
	        }
		} catch (Exception e) {
			log.error(e.getMessage());
		}
    }
}
